package com.atguigu.flink.chapter10.query;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.stream.Stream;

public class TopN {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port",2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        tEnv.executeSql("create table ub(" +
                "user_id bigint," +
                "item_id bigint," +
                "category_id bigint," +
                "behavior string," +
                "ts bigint," +
                "et as to_timestamp_ltz(ts,0)," +
                "watermark for et as et - interval '3' second" +
                ") with(" +
                "  'connector' = 'filesystem', " +
                "  'path' = 'input/UserBehavior.csv', " +
                "  'format' = 'csv' " +
                ")");
        //1、统计在每个窗口的点击量
        Table t1 = tEnv.sqlQuery("select" +
                "item_id,window_start,window_end,count(*) ct " +
                "from table(tumble(table ub,descriptor(et),interval '1' hour))" +
                " where behavior = 'pv' " +
                "group by item_id,window_start,window_end");

        tEnv.createTemporaryView("t1",t1);

        //2、按照点击量在每个窗口进行降序，给每个点击量一个排名
        Table t2 = tEnv.sqlQuery("select " +
                "item_id,window_end,ct, " +
                "row_number() over(partition by window_end order by ct desc) rn " +
                "from t1");

        tEnv.createTemporaryView("t2",t2);


        //3、过滤出topN where rn <= 3
        Table result = tEnv.sqlQuery("select * " +
                "from t2 " +
                "where rn <= 3");

        tEnv.executeSql("create table ub_topn(" +
                "item_id bigint ," +
                " window_end timestamp(3)," +
                "ct bigint," +
                "rn bigint," +
                "primary key(window_end,rn) not enforced" +
                ") with(" +
                " 'connector' = 'jdbc'," +
                " 'url' = 'jdbc:mysql://hadoop162:3306/flink?useSSL=false', " +
                " 'table-name' = 'ub_topn', " +
                " 'username' = 'root', " +
                " 'password' = 'aaaaaa' " +
                ")");

        result.executeInsert("ub_topn");
    }
}




//public class TopN {
//    public static void main(String[] args) {
//        Configuration conf = new Configuration();
//        conf.setInteger("rest.port",2000);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
//        env.setParallelism(1);
//        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
//
//        tEnv.executeSql("create table ub(" +
//                "user_id bigint," +
//                "item_id bigint," +
//                "category_id bigint," +
//                "behavior string," +
//                "ts bigint," +
//                "et as to_timestamp_ltz(ts,0)," +
//                "watermark for et as et interval '3' second  " +
//                ") with(" +
//                "  'connector' = 'filesystem', " +
//                "  'path' = 'input/UserBehavior.csv', " +
//                "  'format' = 'csv' " +
//                ")");
//
//
//        //1、统计每个窗口的点击量
//        Table t1 = tEnv.sqlQuery("select " +
//                "item_id,window_start,window_end,count(*) ct," +
//                "from table(tumble(table ub,desccriptor(et),interval '1' hour) " +
//                "where behavior = 'pv' " +
//                "group by item_id,window_start,window_end");
//
//        tEnv.createTemporaryView("t1",t1);
//
//        //2、按照点击量在每个窗口进行降序，给每个窗口一个排名
//        Table t2 = tEnv.sqlQuery("select " +
//                "itme_id,window_end,ct," +
//                "row_number() over(partition by window_end order by ct desc) rn " +
//                "from t1");
//
//        tEnv.createTemporaryView("t2",t2);
//
//        //3、过滤出topN where rn <= 3
//        Table result = tEnv.sqlQuery("select * " +
//                "from t2 " +
//                "where rn <= 3");
//
//        tEnv.executeSql("create table ub_topn(" +
//        "item_id bigint ," +
//        " window_end timestamp(3)," +
//        "ct bigint," +
//        "rn bigint," +
//        "primary key(window_end,rn) not enforced" +
//        ") with(" +
//        " 'connector' = 'jdbc'," +
//        " 'url' = 'jdbc:mysql://hadoop162:3306/flink?useSSL=false', " +
//        " 'table-name' = 'ub_topn', " +
//        " 'username' = 'root', " +
//        " 'password' = 'aaaaaa' " +
//        ")");
//
//        result.executeInsert("ub_topn");
//
//    }
//}


//public class TopN {
//    public static void main(String[] args) {
//        Configuration conf = new Configuration();
//        conf.setInteger("rest.port",2000);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
//        env.setParallelism(1);
//
//        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
//
//        tEnv.executeSql("create table ub(" +
//                "user_id bigint," +
//                "item_id bigint," +
//                "category_id bigint," +
//                "behavior string," +
//                "ts bigint," +
//                "et as to_timestamp_ltz(ts,0)," +
//                "watermark for et as et - interval '3' second" +
//                ") with(" +
//                "  'connector' = 'filesystem', " +
//                "  'path' = 'input/UserBehavior.csv', " +
//                "  'format' = 'csv' " +
//                ")");
//
//        //1、统计每个窗口的点击量
//        Table t1 = tEnv.sqlQuery("select " +
//                "item_id,window_start,window_end,count(*) ct " +
//                "from table(tumble(table ub,descriptor(et),interval '1' hour)) " +
//                "where behavior = 'pv' " +
//                "group by item_id,window_start,window_end");
//
//        tEnv.createTemporaryView("t1",t1);
//
//
//        //2、按照点击量在每个窗口进行降序，给每个窗口一个排名
//        Table t2 = tEnv.sqlQuery("select " +
//                "item_id,window_end,ct," +
//                "row_number() over(partition by window_end order by ct desc) rn" +
//                "from t1");
//
//        tEnv.createTemporaryView("t2",t2);
//
//
//        //3、过滤出topN where rn <= 3
//        Table result = tEnv.sqlQuery("select * " +
//                "from t2 " +
//                "where rn <= 3");
//
//        tEnv.executeSql("create table ub_topn(" +
//                "item_id bigint," +
//                "window_end timestamp(3)," +
//                "ct bigint," +
//                "rn bigint," +
//                "primary key(window_end,rn) not enforced" +
//                ") with(" +
//                " 'connector' = 'jdbc'," +
//                " 'url' = 'jdbc:mysql://hadoop162:3306/flink?useSSL=false', " +
//                " 'table-name' = 'ub_topn', " +
//                " 'username' = 'root', " +
//                " 'password' = 'aaaaaa' " +
//                ")");
//
//        result.executeInsert("ub_topn");
//
//    }
//}